#include "zab.h"
#include "misc_utils.h"
#include "zab_communication.h"
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#define ZAB_LOG printf

struct NODE_INFO_T
{
    LIST_NODE_T link;
    unsigned int node_id;
    STRING_T* host;
    int election_port;
    int data_port;
    struct sockaddr_in election_addr;
    struct sockaddr_in data_addr;
};

struct ZOO_CONFIG_T
{
    LIST_T nodes;
};


typedef struct VOTE_T
{
    unsigned int node_id;
    ZXID_T zxid;
    unsigned int election_round;
    unsigned int peer_epoch;
    unsigned int leader_id;

    LIST_NODE_T link;

} VOTE_T;

typedef enum LEADER_STATE_T
{
    WAITING_NEW_EPOCH,
    WAITING_NEW_EPOCH_ACK,
    WAITING_NEW_LEADER_ACK,
    LEADER_COMMITED
} LEADER_STATE_T;

typedef struct FOLLOWER_INFO_T
{
    LIST_NODE_T link;
    int sock;
    unsigned int node_id;
    ZAB_PACKETIZER_T packetizer;
    NODE_RUNTIME_T* runtime;
} FOLLOWER_INFO_T;

typedef struct LEADER_PENDING_PROPOSAL_T
{
    LIST_NODE_T link;
    ZXID_T zxid;
    long long timer_id;
    LIST_T received_acks;   // chain of LIST_NODE_T
} LEADER_PENDING_PROPOSAL_T;

struct NODE_RUNTIME_T
{
    ZOO_CONFIG_T* zoo;
    NODE_INFO_T* my_node;
    unsigned int node_id;
    unsigned int current_epoch;
    unsigned int accepted_epoch;    // used in phase 2
    ZXID_T last_zxid;
    NODE_ROLE_T role;

    // config
    STRING_T* runtime_filename;
    // socket stuff
    int election_sock;  // udp socket used for election
    int data_sock;
    long long election_time_id;
    // information for election
    unsigned int election_round;
    unsigned int proposed_leader;
    ZXID_T proposed_zxid;
    unsigned int proposed_epoch;

    LIST_T votes;
    // send buffer
    PACKET_BUFFER_T pb;
    // recv buffer
    ZAB_PACKETIZER_T packetizer;

    // leader data structures
    LEADER_STATE_T leader_state;
    LIST_T followers;
    LIST_T pending_proposals;   // chain of PENDING_PROPOSAL_T
    LEADER_PENDING_PROPOSAL_T new_epoch_acks;
    LEADER_PENDING_PROPOSAL_T new_leader_acks;
    // follower data structures
};
static void zab_load_zoo_config(ZOO_CONFIG_T* cfg, const char* zoo_file);
static void zab_election_start(NODE_RUNTIME_T* runtime, aeEventLoop* loop);
static void zab_election_sock_on_data(struct aeEventLoop *loop, int fd, void *data, int mask);
static int zab_election_on_timer(struct aeEventLoop *loop, long long id, void *data);
static void zab_election_start_timer(struct aeEventLoop* loop, void* data);
static void zab_election_broadcast_vote(NODE_RUNTIME_T* runtime);
static void zab_election_send_vote(NODE_RUNTIME_T* runtime, unsigned int node_id);
static NODE_INFO_T* zab_find_node(ZOO_CONFIG_T* zoo, unsigned int node_id);
static void zab_election_process_vote_msg(NODE_RUNTIME_T* runtime, ZAB_VOTE_MSG_T* msg);
static void zab_election_update_proposal(NODE_RUNTIME_T* runtime, unsigned int leader_id, ZXID_T* zxid, unsigned int epoch);
static VOTE_T* zab_election_find_vote(NODE_RUNTIME_T* runtime, unsigned int node_id);
static void zab_election_set_vote(NODE_RUNTIME_T* runtime, unsigned int node_id,
                                  unsigned int leader, ZXID_T* zxid,
                                  unsigned int election_round, unsigned int epoch);
static void zab_election_clear_votes(NODE_RUNTIME_T* runtime);
static int zab_election_greater(unsigned int a_node_id, unsigned int a_epoch, ZXID_T* a_zxid,
                                unsigned int b_node_id, unsigned int b_epoch, ZXID_T* b_zxid);
static int zab_election_can_finish(NODE_RUNTIME_T* runtime);
static int zab_election_vote_count(NODE_RUNTIME_T* runtime, unsigned int node_id);

// leader logic
static void zab_data_handler(ZAB_PACKET_T* packet, void* data);
static void zab_leader_start(NODE_RUNTIME_T* runtime, aeEventLoop* loop);
static void zab_leader_ack_expire(struct aeEventLoop* loop, long long id, void* data); // one timer for each proposal!
static void zab_leader_heartbeat_expire(struct aeEventLoop* loop, long long id, void* data);
static void zab_leader_on_accept(struct aeEventLoop* loop, int fd, void* data, int mask);
static void zab_leader_on_follower_data(struct aeEventLoop* loop, int fd, void* data, int mask);
static void zab_leader_handle_packet(ZAB_PACKET_T* packet, void* data);
static FOLLOWER_INFO_T* zab_leader_find_follower(NODE_RUNTIME_T* runtime, unsigned int node_id);
static FOLLOWER_INFO_T* zab_leader_find_follower_by_fd(NODE_RUNTIME_T* runtime, int fd);
// pp stands for "pending proposal"
static LEADER_PENDING_PROPOSAL_T* zab_leader_find_pp(NODE_RUNTIME_T* runtime, ZXID_T* zxid);
static void zab_leader_delete_pp(NODE_RUNTIME_T* runtime, LEADER_PENDING_PROPOSAL_T* pp);
static LEADER_PENDING_PROPOSAL_T* zab_leader_find_pp_by_timer(NODE_RUNTIME_T* runtime, long long time_id);
static int zab_leader_got_ack(LEADER_PENDING_PROPOSAL_T* pp, unsigned int acked_node);  // returns acked count

// follower logic
static void zab_follower_start(NODE_RUNTIME_T* runtime);
static void zab_follower_handle_packet(ZAB_PACKET_T* packet, void* runtime);

static void zab_load_zoo_config(ZOO_CONFIG_T* cfg, const char* zoo_file)
{
    FILE* file = fopen(zoo_file, "r");
    if (file != NULL)
    {
        PROP_FILE_T* prop = prop_parse(file);
        int i;

        list_init(&cfg->nodes);
        for (i = 1;; i++)
        {
            NODE_INFO_T node_info;
            memset(&node_info, 0, sizeof(node_info));
            STRING_T* value = prop_get_fmt(prop, "node.%d.id", i);
            if (value != NULL)
            {
                NODE_INFO_T* info;
                node_info.node_id = atoi(value->data);
                value = prop_get_fmt(prop, "node.%d.host", i);
                if (value != NULL)
                    node_info.host = string_new(value->data);
                else
                    break;

                value = prop_get_fmt(prop, "node.%d.election_port", i);
                if (value != NULL)
                    node_info.election_port = atoi(value->data);

                value = prop_get_fmt(prop, "node.%d.data_port", i);
                if (value != NULL)
                    node_info.data_port = atoi(value->data);

                // done
                info = NEW(NODE_INFO_T);
                memcpy(info, &node_info, sizeof(node_info));

                // construct socket addresses
                info->election_addr.sin_family = AF_INET;
                inet_aton(info->host->data, &info->election_addr.sin_addr);
                info->election_addr.sin_port = htons(info->election_port);

                info->data_addr.sin_family = AF_INET;
                info->data_addr.sin_addr = info->election_addr.sin_addr;
                info->data_addr.sin_port = htons(info->data_port);
                // chain
                list_add_tail(&cfg->nodes, &info->link);
            }
            else
                break;
        }
        prop_destroy(prop);
    }
}

NODE_RUNTIME_T* zab_load_node_runtime(const char* zoo_file, const char* runtime_file)
{
    NODE_RUNTIME_T* runtime = NULL;
    FILE* file = fopen(runtime_file, "r");
    if (file != NULL)
    {
        PROP_FILE_T* prop = prop_parse(file);
        STRING_T* value;

        runtime = NEW(NODE_RUNTIME_T);
        runtime->zoo = NEW(ZOO_CONFIG_T);
        if (runtime->zoo != NULL)
        {
            zab_load_zoo_config(runtime->zoo, zoo_file);
        }
        runtime->runtime_filename = string_new(runtime_file);

        value = prop_get(prop, "node_id");
        if (value != NULL)
            runtime->node_id = atoi(value->data);
        value = prop_get(prop, "current_epoch");
        if (value != NULL)
            runtime->current_epoch = atoi(value->data);
        value = prop_get(prop, "accepted_epoch");
        if (value != NULL)
            runtime->accepted_epoch = atoi(value->data);

        prop_destroy(prop);
        fclose(file);
    }
    return runtime;
}

void zab_save_node_runtime(NODE_RUNTIME_T* runtime)
{
    FILE* file = fopen(runtime->runtime_filename->data, "w");
    if (file != NULL)
    {
        fprintf(file, "node_id=%d\n", runtime->node_id);
        fprintf(file, "current_epoch=%d\n", runtime->current_epoch);
        fprintf(file, "accepted_epoch=%d\n", runtime->accepted_epoch);
        fclose(file);
    }
}

int zab_init(NODE_RUNTIME_T* runtime, aeEventLoop* loop)
{
    LIST_NODE_T* node;
    struct sockaddr_in addr;
    ssize_t addr_len;
    int rc;

    // init runtime
    runtime->role = NODE_ELECTING;

    list_init(&runtime->votes);
    packet_buffer_init(&runtime->pb);   // used to construct messages
    // find my node from zoo
    for (node = runtime->zoo->nodes.head; node != NULL; node = node->next)
    {
        NODE_INFO_T* node_info = CONTAINER_PTR(NODE_INFO_T, link, node);
        if (node_info->node_id == runtime->node_id)
        {
            runtime->my_node = node_info;
            break;
        }
    }
    if (runtime->my_node == NULL)
        return 0;

    zab_packetizer_init(&runtime->packetizer, zab_data_handler, runtime);
    zab_election_update_proposal(runtime, runtime->node_id, &runtime->last_zxid, runtime->current_epoch);
    // create socket
    runtime->election_sock = socket(AF_INET, SOCK_DGRAM, 0);
    if (runtime->election_sock < 0)
        return 0;
    memset(&addr, 0, sizeof(addr));
    addr_len = sizeof(addr);
    addr.sin_family = AF_INET;
    addr.sin_port = htons(runtime->my_node->election_port);
    rc = bind(runtime->election_sock, (struct sockaddr*)&addr, addr_len);
    if (rc != 0)
    {
        perror("failed to bind");
        close(runtime->election_sock);
        return 0;
    }
    // add to event loop
    aeCreateFileEvent(loop, runtime->election_sock, AE_READABLE, zab_election_sock_on_data, runtime);

    zab_election_start(runtime, loop);
    return 1;
}

void zab_destroy(NODE_RUNTIME_T* runtime, aeEventLoop* loop)
{
    aeDeleteFileEvent(loop, runtime->election_sock, AE_READABLE);
    aeDeleteTimeEvent(loop, runtime->election_time_id);
    close(runtime->election_sock);

    zab_election_clear_votes(runtime);
    // destroy zoo config
    while (runtime->zoo->nodes.size > 0)
    {
        NODE_INFO_T* node = CONTAINER_PTR(NODE_INFO_T, link, runtime->zoo->nodes.head);
        list_remove_head(&runtime->zoo->nodes);
        string_destroy(node->host);
        DELETE(node);
    }
    DELETE(runtime->zoo);
    string_destroy(runtime->runtime_filename);
}

static void zab_election_start(NODE_RUNTIME_T* runtime, aeEventLoop* loop)
{
    runtime->role = NODE_ELECTING;
    runtime->election_round++;
    // start election...
    zab_election_broadcast_vote(runtime);
    zab_election_start_timer(loop, runtime);
}

static void zab_election_sock_on_data(struct aeEventLoop *loop, int fd, void *data, int mask)
{
    NODE_RUNTIME_T* runtime = (NODE_RUNTIME_T*)data;
    unsigned char buffer[200];
    int len = sizeof(buffer);
    (void)mask;
    (void)loop;
    len = recvfrom(fd, buffer, len, 0, NULL, NULL); // we don't care the source address
    if (len < 200)  // election does not allow big messages
    {
        ZAB_VOTE_MSG_T msg;
        zab_vote_msg_decode(&msg, buffer, len);

        if (runtime->role == NODE_ELECTING)
        {
            zab_election_process_vote_msg(runtime, &msg);
            if ((msg.node_role == NODE_ELECTING) &&
                    (msg.election_round < runtime->election_round))
            {
                zab_election_send_vote(runtime, msg.node_id);
            }
        }
        else
        {
            // tell the peer who is the leader
            if (msg.node_role == NODE_ELECTING)
                zab_election_send_vote(runtime, msg.node_id);
        }
    }
}

static int zab_election_on_timer(struct aeEventLoop *loop, long long id, void *data)
{
    NODE_RUNTIME_T* runtime = (NODE_RUNTIME_T*)data;

    if (runtime->role == NODE_ELECTING)
    {
        // still electing? broadcast vote messages again
        zab_election_broadcast_vote(runtime);
        zab_election_start_timer(loop, runtime);
    }
    aeDeleteTimeEvent(loop, id);
    return 0;
}

static void zab_election_start_timer(struct aeEventLoop* loop, void* data)
{
    NODE_RUNTIME_T* runtime = (NODE_RUNTIME_T*)data;

    runtime->election_time_id = aeCreateTimeEvent(loop, 5000, zab_election_on_timer, data, NULL);
}

static void zab_election_broadcast_vote(NODE_RUNTIME_T* runtime)
{
    LIST_NODE_T* node;
    ZAB_VOTE_MSG_T msg;
    ZAB_PACKET_T* packet;
    // set my information in the msg
    msg.election_round = runtime->election_round;
    msg.node_epoch = runtime->proposed_epoch;
    msg.node_id = runtime->node_id;
    msg.node_role = runtime->role;
    msg.proposed_zxid = runtime->proposed_zxid;
    msg.propsed_leader = runtime->proposed_leader;

    packet = zab_vote_msg_encode(&msg, &runtime->pb);
    ZAB_LOG("Broadcasting vote. Leader=%d\n", runtime->proposed_leader);

    for (node = runtime->zoo->nodes.head; node != NULL; node = node->next)
    {
        NODE_INFO_T* info = CONTAINER_PTR(NODE_INFO_T, link, node);
        // this might be slow?
        if (info->node_id != runtime->node_id)
        {
            sendto(runtime->election_sock,
                   packet, runtime->pb.data_length, 0,
                   (struct sockaddr*)&info->election_addr, sizeof(info->election_addr));
        }
    }
}

static void zab_election_send_vote(NODE_RUNTIME_T* runtime, unsigned int node_id)
{
    NODE_INFO_T* info = zab_find_node(runtime->zoo, node_id);
    ZAB_VOTE_MSG_T msg;
    ZAB_PACKET_T* packet;
    // set my information in the msg
    msg.election_round = runtime->election_round;
    msg.node_epoch = runtime->proposed_epoch;
    msg.node_id = runtime->node_id;
    msg.node_role = runtime->role;
    msg.proposed_zxid = runtime->proposed_zxid;
    msg.propsed_leader = runtime->proposed_leader;

    ZAB_LOG("Sending vote to %d, leader=%d\n", node_id, runtime->proposed_leader);
    packet = zab_vote_msg_encode(&msg, &runtime->pb);
    if (info != NULL)
    {
        sendto(runtime->election_sock,
               packet, runtime->pb.data_length, 0,
               (struct sockaddr*)&info->election_addr, sizeof(info->election_addr));
    }
}

static NODE_INFO_T* zab_find_node(ZOO_CONFIG_T* zoo, unsigned int node_id)
{
    LIST_NODE_T* node;
    for (node = zoo->nodes.head; node != NULL; node = node->next)
    {
        NODE_INFO_T* info = CONTAINER_PTR(NODE_INFO_T, link, node);
        if (info->node_id == node_id)
            return info;
    }
    return NULL;
}

static void zab_election_process_vote_msg(NODE_RUNTIME_T* runtime, ZAB_VOTE_MSG_T* msg)
{
    if (msg->node_role == NODE_ELECTING)
    {
        if (msg->election_round > runtime->election_round)
        {
            // the msg is in a newer round?
            ZAB_LOG("Notification round greater than myself\n");
            runtime->election_round = msg->election_round;
            zab_election_clear_votes(runtime);
            if (zab_election_greater(msg->propsed_leader, msg->node_epoch, &msg->proposed_zxid,
                                     runtime->node_id, runtime->current_epoch, &runtime->last_zxid))
            {
                ZAB_LOG("Notification newer than myself\n");
                zab_election_update_proposal(runtime, msg->propsed_leader, &msg->proposed_zxid, msg->node_epoch);
            }
            else
            {
                // propose myself
                ZAB_LOG("Notification not newer than myself. Proposing myself\n");
                zab_election_update_proposal(runtime, runtime->node_id, &runtime->last_zxid, runtime->current_epoch);
            }
            zab_election_broadcast_vote(runtime);
        }
        else if (msg->election_round < runtime->election_round)
        {
            // I'm newer? ignore
            ZAB_LOG("Notification round %d < mine %d\n", msg->election_round, runtime->election_round);
        }
        else if (zab_election_greater(msg->propsed_leader, msg->node_epoch, &msg->proposed_zxid,
                                      runtime->proposed_leader, runtime->proposed_epoch, &runtime->proposed_zxid))
        {
            ZAB_LOG("Notification greater than my proposal. Updating proposal\n");
            // update proposal
            zab_election_update_proposal(runtime, msg->propsed_leader, &msg->proposed_zxid, msg->node_epoch);
            // broadcast again
            zab_election_broadcast_vote(runtime);
        }

        // update vote
        zab_election_set_vote(runtime, msg->node_id, msg->propsed_leader,
                              &msg->proposed_zxid, msg->election_round, msg->node_epoch);
        if (zab_election_can_finish(runtime))
        {
            if (runtime->proposed_leader == runtime->node_id)
            {
                ZAB_LOG("I am the final leader\n");
                zab_save_node_runtime(runtime);
                runtime->role = NODE_LEADER;
            }
            else
            {
                ZAB_LOG("I am a follower. The final leader is %d\n", runtime->proposed_leader);
                zab_save_node_runtime(runtime);
                runtime->role = NODE_FOLLOWER;
            }
        }
    }
    else if ((msg->node_role == NODE_LEADER) || (msg->node_role == NODE_FOLLOWER))
    {
        // TODO:
        ZAB_LOG("Got notification from %d whose election is done, %d is the leader\n", msg->node_id, msg->propsed_leader);
        zab_election_set_vote(runtime, msg->node_id, msg->propsed_leader,
                              &msg->proposed_zxid, msg->election_round, msg->node_epoch);
        if (msg->election_round == runtime->election_round)
        {
            zab_election_update_proposal(runtime, msg->propsed_leader, &msg->proposed_zxid, msg->node_epoch);
            if (zab_election_can_finish(runtime))
            {
                // we should not be the leader?
                if (runtime->proposed_leader == runtime->node_id)
                {
                    ZAB_LOG("I am the final leader\n");
                    zab_save_node_runtime(runtime);
                    runtime->role = NODE_LEADER;
                }
                else
                {
                    ZAB_LOG("I am a follower. The final leader is %d\n", runtime->proposed_leader);
                    zab_save_node_runtime(runtime);
                    runtime->role = NODE_FOLLOWER;
                }
            }
        }
    }
}

static void zab_election_update_proposal(NODE_RUNTIME_T* runtime, unsigned int leader_id, ZXID_T* zxid, unsigned int epoch)
{
    ZAB_LOG("Updating proposal: leader=%d, zxid=%d:%d, epoch=%d\n", leader_id, zxid->epoch, zxid->count, epoch);
    runtime->proposed_epoch = epoch;
    runtime->proposed_leader = leader_id;
    runtime->proposed_zxid = *zxid;
    zab_election_set_vote(runtime, runtime->node_id, leader_id, zxid, runtime->election_round, epoch);
}

static VOTE_T* zab_election_find_vote(NODE_RUNTIME_T* runtime, unsigned int node_id)
{
    LIST_NODE_T* node;
    for (node = runtime->votes.head; node != NULL; node = node->next)
    {
        VOTE_T* vote = CONTAINER_PTR(VOTE_T, link, node);
        if (vote->node_id == node_id)
            return vote;
    }
    return NULL;
}

static void zab_election_set_vote(NODE_RUNTIME_T* runtime, unsigned int node_id,
                                  unsigned int leader, ZXID_T* zxid,
                                  unsigned int election_round, unsigned int epoch)
{
    VOTE_T* vote = zab_election_find_vote(runtime, node_id);
    if (vote == NULL)
    {
        vote = NEW(VOTE_T);
        list_add_tail(&runtime->votes, &vote->link);
    }
    ZAB_LOG("Set vote: node=%d, leader=%d, round=%d\n", node_id, leader, election_round);
    vote->election_round = election_round;
    vote->node_id = node_id;
    vote->leader_id = leader;
    vote->peer_epoch = epoch;
    vote->zxid = *zxid;
}

static void zab_election_clear_votes(NODE_RUNTIME_T* runtime)
{
    ZAB_LOG("Clearing votes\n");
    while (runtime->votes.size > 0)
    {
        VOTE_T* vote = CONTAINER_PTR(VOTE_T, link, runtime->votes.head);
        list_remove_head(&runtime->votes);
        DELETE(vote);
    }
    zab_election_update_proposal(runtime, runtime->proposed_leader, &runtime->proposed_zxid, runtime->proposed_epoch);
}

static int zab_election_greater(unsigned int a_node_id, unsigned int a_epoch, ZXID_T* a_zxid,
                                unsigned int b_node_id, unsigned int b_epoch, ZXID_T* b_zxid)
{
    if (a_epoch > b_epoch)
        return 1;
    if (a_zxid->epoch > b_zxid->epoch)
        return 1;
    if (a_zxid->count > b_zxid->count)
        return 1;
    if (a_node_id > b_node_id)
        return 1;
    return 0;
}

static int zab_election_can_finish(NODE_RUNTIME_T* runtime)
{
    int count = zab_election_vote_count(runtime, runtime->proposed_leader);
    //ZAB_LOG("Can finish? vote for %d is %d, total %d\n", runtime->proposed_leader, count, runtime->zoo->nodes.size);
    if (count > runtime->zoo->nodes.size / 2)
        return 1;
    return 0;
}

static int zab_election_vote_count(NODE_RUNTIME_T* runtime, unsigned int node_id)
{
    LIST_NODE_T* node;
    int count = 0;
    for (node = runtime->votes.head; node != NULL; node = node->next)
    {
        VOTE_T* vote = CONTAINER_PTR(VOTE_T, link, node);
        if (vote->leader_id == node_id)
        {
            count++;
        }
    }
    return count;
}

///////////////////////////////////////////////////////////////////////////////
// leader logic
///////////////////////////////////////////////////////////////////////////////
static void zab_data_handler(ZAB_PACKET_T* packet, void* data)
{
    NODE_RUNTIME_T* runtime = (NODE_RUNTIME_T*)data;
    if (runtime->role == NODE_FOLLOWER)
        zab_follower_handle_packet(packet, runtime);
}

static void zab_leader_start(NODE_RUNTIME_T* runtime, aeEventLoop* loop)
{
    int rc;
    runtime->data_sock = socket(AF_INET, SOCK_STREAM, 0);
    assert(runtime->data_sock >= 0);

    rc = bind(runtime->data_sock, (struct sockaddr*)&runtime->my_node->data_addr, sizeof(struct sockaddr_in));
    assert(rc == 0);

    rc = listen(runtime->data_sock, 1024);
    assert(rc == 0);

    aeCreateFileEvent(loop, runtime->data_sock, AE_READABLE, zab_leader_on_accept, runtime);

    list_init(&runtime->pending_proposals);
    list_init(&runtime->followers);

    list_init(&runtime->new_epoch_acks.received_acks);
    list_init(&runtime->new_leader_acks.received_acks);
    runtime->leader_state = WAITING_NEW_EPOCH;

    // wait for connection
}
static void zab_leader_stop(NODE_RUNTIME_T* runtime, aeEventLoop* loop)
{
    //
    zab_election_start(runtime, loop);
}

static void zab_leader_ack_expire(struct aeEventLoop* loop, long long id, void* data)
{
    NODE_RUNTIME_T* runtime = (NODE_RUNTIME_T*)data;
    aeDeleteTimeEvent(loop, id);
    // is it ok?
    zab_leader_stop(runtime, loop);
}

static void zab_leader_heartbeat_expire(struct aeEventLoop* loop, long long id, void* data)
{
    // TODO: send heartbeat to all connected peers, using the UDP socket?
}

static void zab_leader_on_accept(struct aeEventLoop* loop, int fd, void* data, int mask)
{
    // new connection
    NODE_RUNTIME_T* runtime = (NODE_RUNTIME_T*)data;
    int sock = accept(fd, NULL, NULL);
    if (sock >= 0)
    {
        FOLLOWER_INFO_T* follower = NEW(FOLLOWER_INFO_T);
        follower->sock = sock;
        follower->node_id = 0;
        follower->runtime = runtime;
        zab_packetizer_init(&follower->packetizer, zab_leader_handle_packet, follower);
        // we don't know the node id, keep it zero now
        list_add_tail(&runtime->followers, &follower->link);
        aeCreateFileEvent(loop, sock, AE_READABLE, zab_leader_on_follower_data, runtime);
        // the follower will send new epoch msg to me.
    }
}

static void zab_leader_on_follower_data(struct aeEventLoop* loop, int fd, void* data, int mask)
{
    NODE_RUNTIME_T* runtime = (NODE_RUNTIME_T*)data;
    FOLLOWER_INFO_T* follower = zab_leader_find_follower_by_fd(runtime, fd);
    unsigned char buffer[1024];
    int len;
    len = recv(fd, buffer, sizeof(buffer), 0);
    if (len > 0)
    {
        zab_packetizer_push(follower, buffer, len);
    }
    else
    {
        // the socket is closed.
        list_remove(&runtime->followers, &follower->link);
        aeDeleteFileEvent(loop, fd, AE_READABLE);
        close(fd);
        DELETE(follower);
        ZAB_LOG("Peer socket closed. Destroying the follower\n");
    }
}

static void zab_leader_handle_packet(ZAB_PACKET_T* packet, void* data)
{
    FOLLOWER_INFO_T* follower = (FOLLOWER_INFO_T*)data;
    NODE_RUNTIME_T* runtime = follower->runtime;
    unsigned short type = ntohs(packet->type);

    if (type == ZAB_PACKET_NEW_EPOCH)
    {
        // TODO:
    }
    else if (type == ZAB_PACKET_NEW_EPOCH_ACK)
    {
        // TODO:
    }
    else if (type == ZAB_PACKET_NEW_LEADER)
    {
        // TODO:
    }
    else if (type == ZAB_PACKET_NEW_LEADER_ACK)
    {
        // TODO:
    }
    // if it's proposal ack
    // if it's data proposal
}

static FOLLOWER_INFO_T* zab_leader_find_follower(NODE_RUNTIME_T* runtime, unsigned int node_id)
{
    // TODO:
    return NULL;
}

static FOLLOWER_INFO_T* zab_leader_find_follower_by_fd(NODE_RUNTIME_T* runtime, int fd)
{
    // TODO:
    return NULL;
}

// pp stands for "pending proposal"
static LEADER_PENDING_PROPOSAL_T* zab_leader_find_pp(NODE_RUNTIME_T* runtime, ZXID_T* zxid);
static void zab_leader_delete_pp(NODE_RUNTIME_T* runtime, LEADER_PENDING_PROPOSAL_T* pp);
static LEADER_PENDING_PROPOSAL_T* zab_leader_find_pp_by_timer(NODE_RUNTIME_T* runtime, long long time_id);
static int zab_leader_got_ack(LEADER_PENDING_PROPOSAL_T* pp, unsigned int acked_node);  // returns acked count

void zab_follower_handle_packet(ZAB_PACKET_T* packet, void* data)
{
    // TODO:
}
